package soendtrs.unused;

import helpers.d;

import java.util.HashMap;
import java.util.Hashtable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

import soenudp.Command;
import soenudp.UDPHelper;
import soenudp.UDPMessage;
import soenudp.UDPException;

/**
 * @author Eman Atallah - 5375606
 * 
 * This is similar the one used by the DTRS but uses locks. The current implementation guarantees
 * the execution of one transaction at a time. This is due to the blocking queue. A more advanced
 * version could make use of locking when dispatching the received transactions in a (callback) 
 * thread. This would require more work and must ensure the order in which clients 
 * transactions are processes.
 * 
 * For example if CustomerID 111333 reserve 15 TOR444 tickets, and another clients reserve 30 
 * tickets 1 second after, CustomerID 111333 must be processes first.
 */

public class DTRSInstance {

    /* Addressing for Multicast MTL Group */
    final String MTL_MULTICAST_ADR  = "235.255.0.1";
    final int    MTL_MULTICAST_PORT = 9200;

    /* Addressing for Multicast TOR Group */
    final String TOR_MULTICAST_ADR  = "235.255.0.2";
    final int    TOR_MULTICAST_PORT = 9100;

    private HashMap<String, Long> ALL_TICKET_COUNT = new HashMap<String, Long>(); // Holds tickets remaining per ShowID.
    private HashMap<String, HashMap<String, Long>> SHOW_RESERVATION = new HashMap<String, HashMap<String, Long>>(); // Holds reservations
    private Hashtable<String, ReentrantLock> LOCK_TABLE = new Hashtable<String, ReentrantLock>(); //  Hold Locks for each ShowID.

    /* Unicast Port of the network main services. */
    final int LEADER_PORT_MTL = 20020; // Change those if you can't bind...
    final int LEADER_PORT_TOR = 20060; // ...the Fron-end will use them to dispatch transactions.
    final int VIEW_UNICT_PORT = 30000; // Unicast port of the GMS's View service.
    
    int leaderPort;            // This one hold the correct port depending on the group type (MTL or TOR).
    int otherLeaderPort;
    int assignedGroupPort;
    String assignedGroupAdr;
    boolean bGMS_Message = false; // Used to skip group Sending on GMS UDP Messages.
    String ourLeaderIP; // Used to warn the Leader (and know about the leader) when a BPing was issued. He only send to group...
    
    int id;                           // ID used in the NodeID Ex: MTL00 + id = MTL001
    boolean isLeader;                 // State variable, Updated by a 'ULEAD' UDPMessage. Caused by the elections.
    boolean validTransaction = false; // Start with false, a valid transaction update this var.
    String nodeID;                    // Verbal ID and the one sent to 'The View'
    String nodeType;                  // MTL, TOR, etc...
    
    UDPHelper multicastGRP = new UDPHelper(); // Used for Group Multicast communications.
    UDPHelper unicast      = new UDPHelper(); // Used in Unicast communicastion: FE, GMS.
    
    @SuppressWarnings({ "unchecked", "rawtypes" })
    public void startDTRS(String type, int id) throws UDPException {
        this.id = id;
        this.nodeType = type;
        this.isLeader = (id > 1) ? false : true;
        this.nodeID   = (id == 999) ?  type + id : type + "00" + id;  

        if (type.equals("MTL")) {
            /* Set proper addressing depending on the DTRS type */
            this.leaderPort        = this.LEADER_PORT_MTL; 
            this.otherLeaderPort   = this.LEADER_PORT_TOR;
            this.assignedGroupPort = this.MTL_MULTICAST_PORT;
            this.assignedGroupAdr  = this.MTL_MULTICAST_ADR;
            
            /* Populate a reservation list for Montreal */
            HashMap MTL101 = new HashMap<String, Long>(); // One Map, holds one show and all 
            HashMap MTL333 = new HashMap<String, Long>(); // associated customer's reservation
            HashMap MTL737 = new HashMap<String, Long>();

            /* Always CUSTOMER ID, TICKETS */
            MTL101.put("311411", 10); // Some reservations to start with...
            MTL333.put("999111", 20);
            MTL737.put("555555", 30);

            ALL_TICKET_COUNT.put("MTL101", (long)  50); // 50  Tickets left for show: MTL101
            ALL_TICKET_COUNT.put("MTL333", (long) 100); // 100 Tickets left for show: MTL333
            ALL_TICKET_COUNT.put("MTL737", (long) 333); // 333 Tickets left for show: MTL737

            SHOW_RESERVATION.put("MTL101", MTL101);
            SHOW_RESERVATION.put("MTL333", MTL333);
            SHOW_RESERVATION.put("MTL737", MTL737);

            LOCK_TABLE.put("MTL101", new ReentrantLock());
            LOCK_TABLE.put("MTL333", new ReentrantLock());
            LOCK_TABLE.put("MTL737", new ReentrantLock());

        } else if (type.equals("TOR")) {
            /* Set proper addressing depending on the DTRS type */
            this.leaderPort        = this.LEADER_PORT_TOR;
            this.otherLeaderPort   = this.LEADER_PORT_MTL;
            this.assignedGroupPort = this.TOR_MULTICAST_PORT;
            this.assignedGroupAdr  = this.TOR_MULTICAST_ADR;
            
            /* Populate a reservation list for Toronto */
            HashMap TOR444 = new HashMap<String, Long>(); // One Map, holds one show and all 
            HashMap TOR888 = new HashMap<String, Long>(); // associated customer's reservation

            /* Always CUSTOMER ID, TICKETS */
            TOR444.put("311411", 10); // Some reservations to start with...
            TOR888.put("555555", 30);

            ALL_TICKET_COUNT.put("TOR444", (long)  50); // 50  Tickets left for show: TOR444
            ALL_TICKET_COUNT.put("TOR888", (long) 100); // 100 Tickets left for show: TOR888

            SHOW_RESERVATION.put("TOR444", TOR444);
            SHOW_RESERVATION.put("TOR888", TOR888);

            LOCK_TABLE.put("TOR444", new ReentrantLock());
            LOCK_TABLE.put("TOR888", new ReentrantLock());
        }

        // Will join the assigned group and will be sending/listending according to the isLeader var.
        multicastGRP.joinGroup(assignedGroupAdr, assignedGroupPort, isLeader, nodeID);    
        
        if (isLeader) {    
            unicast.startUnicastServer(nodeID, leaderPort);  // Used to speak with the Front-End.
            UDPMessage leadersAnnoucement = new UDPMessage();
            leadersAnnoucement.command = Command.NEW_LEADER;
            multicastGRP.sendToGroup(leadersAnnoucement);    // Announce leadership.
        } else {
            // Used to reply at the GMS (The View).
            unicast.startUnicastClient(nodeID); 
        }

        if (id == 999) {
          d.out("PASSIVE DTRS " + nodeID + " waiting for AWAKE Command ") ;
        }
        else {
          d.out("DTRS " + nodeID + " PORT: " + ((isLeader) ? leaderPort + " (LEADER)" : assignedGroupPort + " (GROUP PORT)") 
              + " now ready to process requests"); 
        }

        UDPMessage request = new UDPMessage(); // Will hold one request taken from the FIFO receiving Queue.
        
        /* This is the loop that that consume the FIFO queue...
         * ... First UDP message sent by the Front-End (also Group, GMS, etc...) 
         * ->  First request out to the DTRS */
        while ( (request = consumeQueue(unicast, multicastGRP)) != null )  {
           
            validTransaction = false;
            Command command = request.command; // CHECK, CANCEL, RESERVE, EXCHANGE
            String  showID  = request.show_id; // ShowID
            String  custID  = request.cust_id; // Customer ID
            long    tickets = request.tickets; // Ticket amount

            // d.out("Command: " + command + " CustID: " + custID + " ShowID: " + showID + " Tickets: " + tickets);
           
            String errorMsg = "";
            String successMsg = "";
            long ticketLeft;
            switch (command) {
            case CHECK:
                if (!ALL_TICKET_COUNT.containsKey(showID)) { // Invalid ShowID
                    errorMsg = "Show '" + showID + "' does not exist";
                    validTransaction = false;
                    break;
                } else { // Valid ShowID
                    try {
                        boolean hasLock = LOCK_TABLE.get(showID).tryLock(2, TimeUnit.SECONDS); // Try to acquire the Lock, 2 secs timeout
                        if (!hasLock) {
                            errorMsg = "TIMEOUT, " + showID + " is Locked";
                        } else {
                            d.out("LOCKED    " + showID);
                            ticketLeft = ALL_TICKET_COUNT.get(showID);
                            validTransaction = true;
                            successMsg = "Show '" + showID + "' has '" + ticketLeft + "' tickets available.";
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally { // Make sure we unlock
                        if (LOCK_TABLE.get(showID).isHeldByCurrentThread()) {
                            LOCK_TABLE.get(showID).unlock();
                            d.out("UN-LOCKED " + showID);
                        }
                    }
                }
                break;

            case RESERVE:
                if (!ALL_TICKET_COUNT.containsKey(showID)) { // Invalid ShowID
                    errorMsg = "Show '" + showID + "' does not exist";
                    validTransaction = false;
                    break;
                } else { // Valid ShowID
                    try {
                        boolean hasLock = LOCK_TABLE.get(showID).tryLock(2, TimeUnit.SECONDS); // Try to acquire the Lock, 2 secs timeout
                        if (!hasLock) {
                            errorMsg = "TIMEOUT, " + showID + " is Locked";
                        } else {
                            d.out("LOCKED    " + showID);
                            ticketLeft = ALL_TICKET_COUNT.get(showID);
                            ticketLeft -= tickets;
                            if (ticketLeft > 0) {
                                //ticketLeft = ALL_SHOWS.get(showID);
                                HashMap aReservation = new HashMap<String, Long>();
                                aReservation = SHOW_RESERVATION.get(showID);
                                Long hisTickets; // To hold his current reseravation (how many tickets)
                                if ((hisTickets = (Long) aReservation.get(showID)) != null) // If CustID exist, update it.
                                    aReservation.put(custID, hisTickets + tickets);
                                else // New customer.
                                    aReservation.put(custID, tickets); 

                                SHOW_RESERVATION.put(showID, aReservation); // Updates the reservation list for showID
                                ALL_TICKET_COUNT.put(showID, ticketLeft);   // Updates the ticket count for the showID 

                                successMsg = "You have successfully reserved " + tickets + " tickets for show " + showID;
                                validTransaction = true;

                            } else { // Don't update, not enought tickets ledt
                                errorMsg = "The number of tickets you are trying to reserve is not available. ";
                                errorMsg += "Not enough tickets left.";
                                validTransaction = false;
                            }
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally { // Make sure we unlock
                        if (LOCK_TABLE.get(showID).isHeldByCurrentThread()) {
                            LOCK_TABLE.get(showID).unlock();
                            d.out("UN-LOCKED " + showID);
                        }
                    }
                }
                break;
            case CANCEL:
                if (!ALL_TICKET_COUNT.containsKey(showID)) { // Invalid ShowID
                    errorMsg = "Show '" + showID + "' does not exist";
                    validTransaction = false;
                    break;
                } else { // Valid ShowID
                    try {
                        boolean hasLock = LOCK_TABLE.get(showID).tryLock(2, TimeUnit.SECONDS); // Try to acquire the Lock, 2 secs timeout
                        if (!hasLock) {
                            errorMsg = "TIMEOUT, " + showID + " is Locked";
                        } else {
                            d.out("LOCKED    " + showID);
                            HashMap aReservation = new HashMap<String, Long>();
                            aReservation = SHOW_RESERVATION.get(showID);
                            Long oldNumTick; // To hold his current reseravation (how many tickets)
                            if ((oldNumTick = (Long) aReservation.get(custID)) != null) { // If CustID exist, update it.
                                if (oldNumTick < tickets) { // Customer is trying to cancel more tickets than he have
                                    errorMsg = "You only have " + oldNumTick + " tickets but you are trying to cancel " + tickets
                                            + " tickets. Please try again.";
                                    validTransaction = false;
                                } else { // Update reservation table
                                    aReservation.put(custID, oldNumTick - tickets);
                                    successMsg = "You have successfully canceled " + tickets + " tickets for show " + showID;
                                    validTransaction = true;
                                }
                            } else {
                                errorMsg = "You don't have reservation for the show you are trying to cancel";
                                validTransaction = false;
                                break;
                            }
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally { // Make sure we unlock
                        if (LOCK_TABLE.get(showID).isHeldByCurrentThread()) {
                            LOCK_TABLE.get(showID).unlock();
                            d.out("UN-LOCKED " + showID);
                        }
                    }
                }
                break;
            case EXCHANGE:
                /* Exchange code will go here */
                validTransaction = true;
                successMsg = "You have exchanged your tickets";
                break;
                
            default:
                bGMS_Message = checkForGMSCommands(request, unicast);
                if (bGMS_Message) {
                    validTransaction = true;
                }
                else {
                    validTransaction = false;
                    errorMsg = "Invalid command (" + command + ")";
                }
                break;
            }
           
            if (validTransaction) { // Operation was successful.    
                UDPMessage successReply = new UDPMessage();
                successReply.command = Command.SUCCESS;
                successReply.op_text = successMsg;
                if (isLeader) {   // Only ONE sender (one leader).
                    // Skip 'Pass to Group' for BPING and WAKEUP on Leader's Queue.
                    if (request.command != Command.BPING && request.command != Command.WAKEUP ) { 
                      d.out("Transaction from Front-END Forwarding to Group...");
                      multicastGRP.sendToGroup(request); // Pass to group
                      unicast.sendToNode(request.recv_from_ip, request.recv_from_port, successReply);
                    }
                }
                if (!successMsg.isEmpty() && id != 999) d.out("VALID Transaction! (" + successMsg + ")\n"); 
            } else { // Invalid transaction.
                // Error so we don't pass the request to the group. Replying error to the front-end....
                if (id != 999) d.out("BAD Transaction! (" + errorMsg + ")\n");
                UDPMessage errorReply = new UDPMessage();
                errorReply.command = Command.ERROR;
                errorReply.op_text = errorMsg;
                if (isLeader) {
                    unicast.sendToNode(request.recv_from_ip, request.recv_from_port, errorReply);
                }
            }
            if (this.isLeader && request.command == Command.WAKEUP) id = 1;
        } /* Back to the top of while() */
    }
    
    /* 
     * Used to check if the GMS is looking to contact the group.
     * Most important command are BPING, ULEAD, and WAKEUP
     */
    private boolean checkForGMSCommands(UDPMessage request, UDPHelper quick) throws UDPException {
        boolean isValidCommand = false;
        UDPMessage response = new UDPMessage();
        switch (request.command) {
        
        case WAKEUP: // A passive node is about to be awaken.
            if (id == 999) { // The passive node is id=999
                d.out("GMS sent the following: " + request.command);  
                d.out("--\n" + request);
                response.command = Command.WAKEUP;
                boolean asLeader = (request.op_text).equals("LEADER");
                awakePassiveNode(777, asLeader); // true/false -> isLeader.
                response.op_text = this.nodeID; // Reply back our new nodeID
                quick.sendToNode(request.recv_from_ip, VIEW_UNICT_PORT, response);
            }
            isValidCommand = true;
            break;  
            
        case BPING:
             if (id != 999) {
               d.out("GMS sent the following: " + request.command);
               response.command = Command.PONG;
               response.op_text = nodeID;
               quick.sendToNode(request.recv_from_ip, VIEW_UNICT_PORT, response);
            }
            isValidCommand = true;
            break;
            
        case WHO_LEADS: // GMS is requesting our leader's IP Address.
            if (id != 999) {
              d.out("GMS sent the following message: " + request.command);
              response.command = Command.WHO_LEADS;
              d.out("GMS> " + ourLeaderIP);
              response.op_text = nodeType + "--" + ourLeaderIP;        
              quick.sendToNode(request.recv_from_ip, VIEW_UNICT_PORT, response);
            }
            isValidCommand = true;
            break;
            
        case NEW_LEADER: // Not really from GMS but usefull to know the leader's new IP after the elections.
            if (id != 999) d.out("LEADER sent the following message: " + request.command);
           
            ourLeaderIP = request.recv_from_ip;
            d.out("LEADER< " + ourLeaderIP);
            isValidCommand = true;
            break;
        }
        return isValidCommand;
    }

    /* Method used by the passive that makes her switch to either a replica or a leader) */
    private void awakePassiveNode(int theID, boolean toBeLeader) throws UDPException {
        this.id = theID;
        this.nodeID = nodeID.replace("999", String.valueOf(theID)); // Will becode MTL777 on 'The View'.

        if (toBeLeader) { // Make this node a Leader.
          d.out("-------> NEW LEADER NODE <-------");
          unicast.stopUnicast();     // Stop unicast services, will now use a specefic  port to receive.
          multicastGRP.leaveGroup(); // Must leave the group 1st to switch from listener to broadcaster.
          UDPMessage leadersAnnoucement = new UDPMessage();
          leadersAnnoucement.command = Command.NEW_LEADER;  // New leadership will broadcast new IP Adr.
          multicastGRP.joinGroup(assignedGroupAdr, assignedGroupPort, true, nodeID); // Join as a Leader
          unicast.startUnicastServer(nodeID, leaderPort);   // Start the unicast server to speak with FE
          multicastGRP.sendToGroup(leadersAnnoucement);
          this.nodeID = nodeID.replace("777", "001"); // Make nodeID = 001, this is the one use by leader.
          this.isLeader = true;
        }
        d.out("New NodeID: " + nodeID); // Print out the new NodeID on the console.
    }
    
    /* Method used by replica to determine if the message is from their leader */
    private boolean isFromOurLeader(UDPMessage request) {
      if (request.command == Command.CHECK   || 
          request.command == Command.RESERVE || 
          request.command == Command.CANCEL) {
            return true;
      } else {
            return false;
      }   
    }
    
    /*
     * Consume the appropriated Queue based on the isLeader variable
     */
    private UDPMessage consumeQueue(UDPHelper unicastFE, UDPHelper multicastGRP) {
        try {
            if (isLeader) { // Get message from the unicast Queue (Front End send to this one).
                return  unicastFE.getMessage();
            } else { // Messages from the group's receceiving Queue. (Leader send th thos one).
                UDPMessage request =  multicastGRP.getMsgFromGroup();
                if ( isFromOurLeader(request) ) {
                    ourLeaderIP = request.recv_from_ip; // Keeps track of the IP, in case its changes (elections).
                }
                return request;
            }
        } catch (UDPException e) {
            e.printStackTrace();
        }
        return null;
    }
}